/*
 *
 */

package columnar;

import bitmap.*;
import btree.*;
import bufmgr.*;
import diskmgr.*;
import global.*;

import iterator.*;

import java.io.*;

import heap.*;

import java.lang.reflect.Array;
import java.util.ArrayList;
import java.util.List;

/**
 * 
 * @author SHIVANI
 */

interface Filetype {
	int TEMP = 0;
	int ORDINARY = 1;

} // end of Filetype

public class Columnarfile extends Object implements Filetype, GlobalConst {
	private static short REC_LEN_STRING = 160;
	private static short REC_LEN_INT = 32;

	PageId _firstDirPageId; // page number of header page
	int _ftype;
	private boolean _columnarFile_deleted;
	private String _columnarFileName;
	private static int tempColumnarFilecount = 0;
	private int numColumns;
	private AttrType[] type;
	private Heapfile[] columnArray;
	private String[] _columnNames;
	private Heapfile delHeapFile;
    private Heapfile indexHeapFile;
    private List<String> indexFile = new ArrayList<String>();
	/**
	 * Constructor
	 */
	public Columnarfile(String name, int numColumns, AttrType[] type,
			String[] columnNames) throws HFException, HFBufMgrException,
			HFDiskMgrException, IOException {
		this.numColumns = numColumns;
		this.type = type;
		_columnNames = columnNames;

		columnArray = new Heapfile[numColumns + 1];

		_columnarFile_deleted = true;
		_columnarFileName = null;

		if (name == null) {
			_columnarFileName = "ColumnarFile";
			_ftype = TEMP;
			tempColumnarFilecount++;
		} else {
			_columnarFileName = name;
			_ftype = ORDINARY;
		}

		// create the columnarfile.hdr heapfile
		Heapfile hdrHeapFile = null;
		try {
			hdrHeapFile = new Heapfile(_columnarFileName + ".hdr");

			System.out.println("*** Heap file for a column hdr created. \n");
		} catch (Exception e) {
			System.out
					.println("*** Could not create heap file for a column hdr \n");
			e.printStackTrace();
		}
		

		try {
			delHeapFile = new Heapfile(_columnarFileName + ".del");

			System.out.println("*** Heap file for a column delete created. \n");
		} catch (Exception e) {
			System.out
					.println("*** Could not create heap file for a column hdr \n");
			e.printStackTrace();
		}

       try {
			indexHeapFile = new Heapfile(_columnarFileName + ".index");

			System.out.println("*** Heap file for a column index created. \n");
		} catch (Exception e) {
			System.out
					.println("*** Could not create heap file for a column hdr \n");
			e.printStackTrace();
		}


                // create heapfiles for every column.
		for (int i = 0; i < numColumns; i++)
                {
			if (_columnNames[i] == null || _columnNames[i] == "") {
				_columnNames[i] = String.valueOf(i);
			}

			String columnHeapfileName = _columnarFileName + "." + i;

			Heapfile tempHeapFile = null;
			try {
				tempHeapFile = new Heapfile(columnHeapfileName);
				columnArray[i] = tempHeapFile;
				// create Tuple for every columnar file and inser it in the
				// heapfile.

				Tuple hdrTuple = new Tuple();
				AttrType[] hdrtype = new AttrType[4];
				hdrtype[0] = new AttrType(AttrType.attrInteger);
				hdrtype[1] = new AttrType(AttrType.attrString);
				hdrtype[2] = new AttrType(AttrType.attrInteger);
				hdrtype[3] = new AttrType(AttrType.attrString);

				hdrTuple.setHdr((short) 4, hdrtype, new short[] { REC_LEN_INT,
						REC_LEN_STRING, REC_LEN_INT, REC_LEN_STRING });
				hdrTuple.setIntFld(1, i);
				hdrTuple.setIntFld(3, type[i].attrType);
				hdrTuple.setStrFld(2, columnHeapfileName);
				hdrTuple.setStrFld(4, _columnNames[i]);

				hdrHeapFile.insertRecord(hdrTuple.getTupleByteArray());

				System.out.println("*** Heap file for a column:"
						+ columnHeapfileName + " created. \n");
			} catch (Exception e) {
				System.err
						.println("*** Could not create heap file for a column:"
								+ columnHeapfileName + " \n");
				e.printStackTrace();
			}
		}

		columnArray[numColumns] = hdrHeapFile;
		_columnarFile_deleted = false;

	} // end of constructor

	// Read DB from disk
	public Columnarfile(String name) throws HFException, HFBufMgrException,
			HFDiskMgrException, IOException, Exception {
		RID rid = new RID();
		Tuple aTuple = new Tuple();
		Heapfile headerFile = new Heapfile(name + ".hdr");
		_columnarFileName = name;
		
		//read delete heapfile
		delHeapFile = new Heapfile(name + ".del");

        //read Index heapfile
		indexHeapFile = new Heapfile(name + ".index");

		numColumns = headerFile.getRecCnt();

		columnArray = new Heapfile[numColumns + 1];
		_columnNames = new String[numColumns];
		type = new AttrType[numColumns];

		Scan scan = headerFile.openScan();

		for (int i = 0; i < numColumns; i++) {
			aTuple = scan.getNext(rid);
			AttrType[] hdrtype = new AttrType[4];
			hdrtype[0] = new AttrType(AttrType.attrInteger);
			hdrtype[1] = new AttrType(AttrType.attrString);
			hdrtype[2] = new AttrType(AttrType.attrInteger);
			hdrtype[3] = new AttrType(AttrType.attrString);
			aTuple.setHdr((short) 4, hdrtype, new short[] { REC_LEN_INT,
					REC_LEN_STRING, REC_LEN_INT, REC_LEN_STRING });

			int columnNo = aTuple.getIntFld(1);
			String columnFileName = aTuple.getStrFld(2);
			int columnType = aTuple.getIntFld(3);
			String columName = aTuple.getStrFld(4);

			columnArray[i] = new Heapfile(columnFileName);
			type[i] = new AttrType(columnType);
			_columnNames[i] = columName;
		}

		columnArray[numColumns] = headerFile;
		_columnarFileName = name;
		scan.closescan();
	}

	public void deleteColumnarFile() throws HFException {
		Heapfile columnHeapFiles;
		for (int i = 0; i < columnArray.length; i++) {
			columnHeapFiles = columnArray[i];
			if (columnHeapFiles != null) {
				try {
					columnHeapFiles.deleteFile();
				} catch (Exception e) {
					throw new HFException(e, "ColumnFile: delete file failed");
				}
			}
		}
		columnHeapFiles = null;
		columnArray = null;
	}

        private int recCount = 0;
	public TID insertTuple(Tuple dataTuple) throws InvalidSlotNumberException,
			InvalidTupleSizeException, SpaceNotAvailableException, HFException,
			HFBufMgrException, HFDiskMgrException, IOException, Exception {
		//int recCount = this.getTupleCnt();
		RID[] rids = new RID[numColumns];
		RID rid = new RID();

		for (int i = 0; i < numColumns; i++)
                {
                    Heapfile hf = columnArray[i];
                    //recCount = hf.getRecCnt();
                    Tuple entry = new Tuple();

                    try {
			AttrType[] typeTemp = new AttrType[1];
			if (type[i].attrType == AttrType.attrString) {
				typeTemp[0] = new AttrType(AttrType.attrString);
				entry.setHdr((short) 1, typeTemp,
                        			new short[] { REC_LEN_STRING });
				String strData = dataTuple.getStrFld(i + 1);
				entry.setStrFld(1, strData);
			} else if (type[i].attrType == AttrType.attrInteger) {
				typeTemp[0] = new AttrType(AttrType.attrInteger);
				entry.setHdr((short) 1, typeTemp,
        					new short[] { REC_LEN_INT });
				int intData = dataTuple.getIntFld(i + 1);
				entry.setIntFld(1, intData);
                	}
			rids[i] = hf.insertRecord(entry.getTupleByteArray());
                    } catch (Exception e) {
                        System.err.println("*** Could not insert the tuple data."
                                		+ recCount + "\n");
			e.printStackTrace();
			Runtime.getRuntime().exit(1);
                    }
                }

                // Create the TID to return.
		TID tid = new TID(numColumns);
		tid.setNumRID(numColumns);
		tid.setRIDs(rids);
		tid.setPosition(++recCount); // the Position Id starts with 1 and not 0;
		
		return tid;
	}

	public Tuple getTuple(TID tid) throws InvalidSlotNumberException,
			InvalidTupleSizeException, HFException, HFDiskMgrException,
			HFBufMgrException, Exception {
		Tuple returnTuple = new Tuple();
		Tuple tempTuple = new Tuple();
		AttrType[] type = new AttrType[1];

		for (int i = 0; i < numColumns; i++) {
			tempTuple = columnArray[i].getRecord(tid.recordIDs[i]);
			type[0] = new AttrType(type[i].attrType);
			try {
				if (type[i].attrType == AttrType.attrString) {
					returnTuple.setHdr((short) 1, type, new short[] { REC_LEN_STRING });
					String strData = tempTuple.getStrFld(1);
					returnTuple.setStrFld(i + 1, strData);
				} else if (type[i].attrType == AttrType.attrInteger) {
					returnTuple.setHdr((short) 1, type, new short[] { REC_LEN_INT });
					int intData = tempTuple.getIntFld(1);
					returnTuple.setIntFld(i + 1, intData);
				}
			} catch (Exception e) {
				System.err.println("*** Could not get tuple. \n");
			}
		}

		return returnTuple;

	}

	public ValueClass getValue(TID tid, int column)
			throws InvalidSlotNumberException, InvalidTupleSizeException,
			HFException, HFDiskMgrException, HFBufMgrException, Exception {
		ValueClass columnVal = null;
		RID rid = tid.recordIDs[column];
		Tuple tempTuple = columnArray[column].getRecord(rid);

		try {
			if (type[column].attrType == AttrType.attrString) {
				String strData = tempTuple.getStrFld(column + 1);
				StringValue strVal = new StringValue(strData);
				return strVal;
			} else if (type[column].attrType == AttrType.attrInteger) {
				int intData = tempTuple.getIntFld(column + 1);
				IntegerValue intVal = new IntegerValue(intData);
				return intVal;
			}
		} catch (Exception e) {
			System.err.println("*** Could not get the value out of the column\n");
		}
		return columnVal;
	}

	public int getTupleCnt() throws InvalidSlotNumberException,
			InvalidTupleSizeException, HFDiskMgrException, HFBufMgrException,
			IOException {
		int answer = 0;
		answer = columnArray[0].getRecCnt();
		return answer;
	} // end of getTupleCnt

	public boolean updateTuple(TID tid, Tuple newtuple) throws Exception {
		boolean result = false;
		Tuple returnTuple = new Tuple();
		for (int i = 0; i < numColumns; i++) {
			try {
				AttrType[] typeTemp = new AttrType[1];
				if (type[i].attrType == AttrType.attrString) {
					typeTemp[0] = new AttrType(AttrType.attrString);
					returnTuple.setHdr((short) 1, typeTemp,
							new short[] { REC_LEN_STRING });
					String strData = newtuple.getStrFld(i + 1);
					returnTuple.setStrFld(1, strData);
				} else if (type[i].attrType == AttrType.attrInteger) {
					typeTemp[0] = new AttrType(AttrType.attrInteger);
					returnTuple.setHdr((short) 1, typeTemp,
							new short[] { REC_LEN_INT });
					int intData = newtuple.getIntFld(i + 1);
					returnTuple.setIntFld(1, intData);
				}
			} catch (Exception e) {
				System.err.println("*** Could not update a tuple. \n");
			}
			result = columnArray[i].updateRecord(tid.recordIDs[i], returnTuple);
		}

		return result;
	}

	public boolean updateColumnofTuple(TID tid, Tuple newtuple, int column)
			throws Exception {
		boolean result = false;

		result = columnArray[column].updateRecord(tid.recordIDs[column],
				newtuple);

		return result;
	}

	public Scan openColumnScan(int columnNo) throws InvalidTupleSizeException,
			IOException {
		Scan newColumnScan = columnArray[columnNo].openScan();
		return newColumnScan;
	}

	public TupleScan openTupleScan() throws InvalidTupleSizeException,
			IOException {

		TupleScan newTupleScan = new TupleScan(this);
		return newTupleScan;
	}

	public boolean createBTreeIndex(String columnname)
		throws Exception{
		boolean resultVal = true;
		String filename = getColumnFileName(columnname) + ".btree";
		Scan scan=null;
		BTreeFile btf=null;
		int keysize;
		RID rid = new RID();
		int column;
		column=this.getColumnOffset(columnname)-1;
		int count=0;

		if (type[column].attrType == AttrType.attrInteger) {
			keysize = REC_LEN_INT;
		} else if (type[column].attrType == AttrType.attrString) {
			keysize = REC_LEN_STRING;
		} else {
			System.err.println("*** Invalid key type.\n");
			return false;
		}
		try{
			btf=new BTreeFile(filename);
			resultVal = true;;
			System.out.println(filename+" has already existed.");
			btf.close();
			return resultVal;
		}catch(Exception e)
		{
			btf = new BTreeFile(filename, type[column].attrType, keysize,
					DeleteFashion.FULL_DELETE);// or DeleteFashion.NAIVE_DELETE

			Tuple temp = null;

			scan = openColumnScan(column);
		
			while((temp =  scan.getNext(rid)) != null) 
			{
				Columnarfile.SetOneColumnTupleHeader(temp, type[column]);
			
				if (type[column].attrType == AttrType.attrInteger) 
				{
					int key = 0;
					key = temp.getIntFld(1);
					btf.Insert(new IntegerKey(key), rid);
				}
				else if (type[column].attrType == AttrType.attrString)
				{
					String key = "";
					key = temp.getStrFld(1);
					btf.Insert(new StringKey(key), rid);
				}
				count++;
			}
			scan.closescan();
			BT.printBTree(btf.getHeaderPage());
			btf.close();
		}
		return resultVal;
	}


	public boolean createBitMapIndex(int columnNo, ValueClass value) {
		boolean resultVal = false;
		String filename = _columnarFileName + "." + columnNo + "." + "bitmap"+"."+value.toString();
		try {
			BitMapFile bmf=new BitMapFile(filename);
			resultVal = true;
			System.out.println(filename+" has already existed.");
			bmf.close();
			return resultVal;
		} catch (Exception e) {
			try {
				new BitMapFile(filename, this, columnNo, value);
				resultVal = true;
                                insertIndexName(filename);  //Add index entry to the index heapfile.
			} catch (Exception e1) {
				System.err.println("*** Could not create a BitMap file. \n");
			}
		}

		return resultVal;
	}

	public String columnarFileName() {
		return _columnarFileName;
	}

	public AttrType[] getAttrTypes() {
		return type;
	}

	public Heapfile[] ColumnArray() {
		return columnArray;
	}

	public int NumberOfColumns() {
		return columnArray.length - 1;
	}

	public static void SetOneColumnTupleHeader(Tuple tuple, AttrType type)
			throws InvalidTypeException, InvalidTupleSizeException, IOException {
		AttrType[] typeTemp = new AttrType[1];

		switch (type.attrType) {
		case AttrType.attrInteger:
			typeTemp[0] = new AttrType(AttrType.attrInteger);
			tuple.setHdr((short) 1, typeTemp, new short[] { 32 });
			break;
		case AttrType.attrString:
			typeTemp[0] = new AttrType(AttrType.attrString);
			tuple.setHdr((short) 1, typeTemp, new short[] { 160 });
		}

	}

	public String getColumnFileName(String columnName) {
		String columnFileName = "";
		for (int i = 0; i < numColumns; i++) {
			if (_columnNames[i].equals(columnName)) {
				columnFileName = _columnarFileName + "." + i;
			}
		}

		return columnFileName;
	}
	
	public AttrType getColumnType(String columnName) {
		for (int i = 0; i < numColumns; i++) {
			if (_columnNames[i].equals(columnName)) {
				return type[i];
			}
		}
		return new AttrType(AttrType.attrNull);
	}
	
	public int getColumnOffset(String columnName) {
		for (int i = 0; i < numColumns; i++) {
			if (_columnNames[i].equals(columnName)) {
				return i+1;
			}
		}
		return 0;
	}
	
	//DUMMY METHOD - needs to be implemented specifying sizes of strings
	public short[] getSizesOfFileds()
	{
		short[] sizes = new short[NumberOfColumns()];
		for(int i = 0; i<NumberOfColumns();i++)
		{
			if (type[i].attrType == AttrType.attrString)
				sizes[i] = REC_LEN_STRING;
			else if(type[i].attrType == AttrType.attrInteger)
				sizes[i] = REC_LEN_INT;
		}
		return sizes;
	}

	public boolean purgeAllDeletedTuples() throws Exception {
		  boolean result = false;

            RID rid = new RID();
            TID tid = new TID(numColumns);
            List<TID> tidList = new ArrayList<TID>();
            Tuple tuple = new Tuple();
            int position = 0;
            int curPosition = 0;
            TupleScan scan = new TupleScan(this);
           
            Sort sort = startSortDelMethod();
            Tuple t = null;
            while ((t = sort.get_next()) != null)
            {
                position = t.getIntFld(1);

                for(int i=curPosition;i<position;i++)
                {
                    t = scan.getNext(tid);
                    curPosition++;
                }
                //tidList.add(tid);
                
               // TID delTid = tidList.get(j);
                for (int i = 0; i < numColumns; i++)
                {
                   result = columnArray[i].deleteRecord(tid.recordIDs[i]);
                }
            }
            scan.closeTuplescan();
            sort.close();
            
            Heapfile delHeap = new Heapfile(_columnarFileName+".del");
            delHeap.deleteFile();
            
            SystemDefs.JavabaseBM.flushAllPages();
            
            delHeapFile= new Heapfile(_columnarFileName+".del");
            
            
            if(result == true)
                   System.out.println("Tuple Delete Success.");

            return result;
		 }
	
        public Sort startSortDelMethod() throws Exception
        {
            AttrType[] attrType = new AttrType[1];
                attrType[0] = new AttrType(AttrType.attrInteger);
                short[] attrSize = new short[1];
                attrSize[0] = REC_LEN_INT;
                FldSpec[] projlist = new FldSpec[1];
                RelSpec rel = new RelSpec(RelSpec.outer);
                projlist[0] = new FldSpec(rel, 1);

                TupleOrder[] order = new TupleOrder[2];
                order[0] = new TupleOrder(TupleOrder.Ascending);
                order[1] = new TupleOrder(TupleOrder.Descending);

                FileScan fscan = new FileScan(_columnarFileName+".del", attrType, attrSize, (short) 1, 1, projlist, null);
                
                Sort sort = new Sort(attrType, (short) 1, attrSize, fscan, 1, order[0], REC_LEN_INT, 12);

                return sort;
        }

        
        public void InsertDeletedPosition(int position) throws InvalidTypeException, InvalidTupleSizeException, IOException, FieldNumberOutOfBoundException, InvalidSlotNumberException, SpaceNotAvailableException, HFException, HFBufMgrException, HFDiskMgrException
        {    	
    		Tuple delTuple = new Tuple();
    		Columnarfile.SetOneColumnTupleHeader(delTuple, new AttrType(AttrType.attrInteger));
    		delTuple.setIntFld(1, position);
			delHeapFile.insertRecord(delTuple.getTupleByteArray());
        }
     
        
        public Tuple getCompleteTupleByPosition(int position) throws Exception
        {    	
    		Tuple returnTuple = new Tuple();
    		returnTuple.setHdr((short)NumberOfColumns(), type, getSizesOfFileds());
    		
    		for (int i = 0; i < numColumns; i++)
            {
                Heapfile hf = columnArray[i];
	            Tuple tempTuple = new Tuple();
	            tempTuple = hf.getTupleForGivenPosition(position-1);
	            
				if (type[i].attrType == AttrType.attrString) {
					tempTuple.setHdr((short) 1, new  AttrType[]{new AttrType(AttrType.attrString)}, new short[] { REC_LEN_STRING });
					String strData = tempTuple.getStrFld(1);
					returnTuple.setStrFld(i + 1, strData);
				} else if (type[i].attrType == AttrType.attrInteger) {
					tempTuple.setHdr((short) 1, new  AttrType[]{new AttrType(AttrType.attrInteger)}, new short[] { REC_LEN_INT });
					int intData = tempTuple.getIntFld(1);
					returnTuple.setIntFld(i + 1, intData);
				}
            }

            return returnTuple;
        }
        
        
        public String getColumnFileNameByNo(int columnNo)
        {
            String columnFileName = "";
            for (int i = 0; i < numColumns; i++) {
                if (i == columnNo) {
                    columnFileName = _columnNames[i];//_columnarFileName + "." + i;
                }
            }

            return columnFileName;
	}

        public void insertIndexName(String iName) throws InvalidTypeException, InvalidTupleSizeException, IOException, FieldNumberOutOfBoundException, InvalidSlotNumberException, SpaceNotAvailableException, HFException, HFBufMgrException, HFDiskMgrException
        {
    		Tuple indexTuple = new Tuple();
    		Columnarfile.SetOneColumnTupleHeader(indexTuple, new AttrType(AttrType.attrString));
    		indexTuple.setStrFld(1, iName);
    		indexHeapFile.insertRecord(indexTuple.getTupleByteArray());
            indexFile.add(iName);
        }

        public void createBitmapForColumn(int columnNo)
                throws Exception
        {
            //Get all Bitmap_index in a list
            Tuple ituple = new Tuple();
            RID irid = new RID();
            Scan iscan = indexHeapFile.openScan();
            indexFile = new ArrayList<String>();
            while ((ituple = iscan.getNext(irid)) != null)
            {
                Columnarfile.SetOneColumnTupleHeader(ituple, new AttrType(AttrType.attrString));
                indexFile.add(ituple.getStrFld(1));
            }
            iscan.closescan();

        try {
            Scan scan = openColumnScan(columnNo);
            RID rid = new RID();
            Tuple t = new Tuple();
            AttrType atype = getColumnType(getColumnFileNameByNo(columnNo));

            while((t =  scan.getNext(rid)) != null)
            {
                Columnarfile.SetOneColumnTupleHeader(t,atype);
                ValueClass value = null;
                if(atype.attrType == AttrType.attrInteger)
                {
                    value = new StringValue(t.getIntFld(1)+"");
                }
                else if(atype.attrType == AttrType.attrString)
                {
                    value = new StringValue(t.getStrFld(1));
                }

                String bmFileName = _columnarFileName + "." + columnNo + "." + "bitmap"+"."+value.toString();
                if(indexFile.contains(bmFileName) != true)
                    createBitMapIndex(columnNo, value);   //cf.getColumnOffset(columnName)-1
                else
                    System.out.println("Bitmap for "+getColumnFileNameByNo(columnNo)+", value "+value+" already exist. ");
            }
            scan.closescan();
        }
        catch (Exception e) {
            System.err.println("Bitmaps Not created");//e.toString());
            Runtime.getRuntime().exit(1);
        }
    }


    public List indexFileList() throws Exception {
        if(indexFile.isEmpty())
        {
            //Get all Bitmap_index in a list
            Tuple ituple = new Tuple();
            RID irid = new RID();
            Scan iscan = indexHeapFile.openScan();
            indexFile = new ArrayList<String>();
            while ((ituple = iscan.getNext(irid)) != null)
            {
                Columnarfile.SetOneColumnTupleHeader(ituple, new AttrType(AttrType.attrString));
                indexFile.add(ituple.getStrFld(1));
            }
            iscan.closescan();
            return indexFile;
        }else
            return indexFile;
    }
}// End of Coulmnarfile.
